package NBA

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import scala.io.Source


object NBADataClear {
  def main(args: Array[String]): Unit = {
    // 实例化
    val conf = new SparkConf().setAppName("NBADataClear").setMaster("local")
    val sc = new SparkContext(conf)
    
    var mapping:Map[String,String] = Map()  // 声明Map存储新旧队名的映射关系
    var eastList:List[String] =List()  // 声明List存储东区队名
    var westList:List[String] =List()  // 声明List存储西区队名
    
    val source_map = Source.fromFile("A:/input/mappings.txt")
    val source_east = Source.fromFile("A:/input/east.txt")
    val source_west = Source.fromFile("A:/input/west.txt")
    
    // 读取文件中的内容初始化maping eastList westList
    
    for(line <- source_map.getLines()){
      val arr = line.split(",")
      mapping +=(arr(0) -> arr(1))
    }
    for(line <- source_east.getLines()){
      eastList = line :: eastList
    }
    for(line <- source_west.getLines()){
      westList = line :: westList
    }
    
    // 释放资源
    source_map.close()
    source_east.close()
    source_west.close()
    
    // 创建广播变量
    val bc_map = sc.broadcast(mapping)
    val bc_east = sc.broadcast(eastList)
    val bc_west = sc.broadcast(westList)
    
    // 读取文件进行清洗
    sc.textFile("A:/input/nba.csv")
    .map(_.split(","))
    .map(arr => {
      val first = arr(2)  // 冠军队名
      val second = arr(4)  // 亚军队名
      
      // 根据数据文件中的队名获取新队名
      val new_first = bc_map.value.getOrElse(first, first)
      val new_second = bc_map.value.getOrElse(second, second)
      
      var result:Array[String] = Array()
      result = result:+arr(0):+arr(1):+new_first:+arr(3):+new_second
      
      // 获取冠军球队的分区信息
      val year = arr(0).toInt
      if(year >= 1970){
        var flag = "E"
        if(bc_west.value.contains(new_first)){
          flag = "W"
        }
        result = result :+ flag
      }
      result.mkString(",")
    })
    .foreach(println)
//    .repartition(1)
//    .saveAsTextFile("A:/output3/nba_clear")
    
  }
}